#include "util/io.hxx"
#include "util/fs.hxx"
#include <assert.h>
#include <sys/mman.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <iostream>


PosixSequentialReader::PosixSequentialReader(const std::string& fname)
: SequentialReader(fname) {

}

PosixSequentialReader::~PosixSequentialReader() {

	if (m_file) {
        PosixSequentialReader::close();
	}
}


DbStatus PosixSequentialReader::init() {

	DbStatus ret;

	m_file = fopen(m_filename.c_str(), "r"); 
	if(m_file == NULL) ret = DbStatus(m_filename, errno);

	return ret;

}

DbStatus PosixSequentialReader::read(uint32_t n, std::string& result) {

	DbStatus ret;

	result.assign(n, 0);

	uint32_t r = fread((char*)result.c_str(), 1, n, m_file);
	if(r < n) {
	
		if(feof(m_file)) {
        
            result.resize(r);
        }
		else ret = DbStatus(m_filename, errno);
	}

	return ret;
}


DbStatus PosixSequentialReader::skip(uint64_t n) {

	DbStatus ret;
	if(fseek(m_file, n, SEEK_CUR) < 0) ret = DbStatus(m_filename, errno);

	return ret;
}

DbStatus PosixSequentialReader::getCurPos(uint64_t& pos) {

    DbStatus ret;
    if( (pos = ftell(m_file)) < 0) ret = DbStatus(m_filename, errno);

    return ret;
}

DbStatus PosixSequentialReader::close() {

    DbStatus ret;

    if(fclose(m_file)) ret = DbStatus("osixSequentialReader::close", errno);

    return ret;
}


uint32_t PosixMmapSequentialWriter::rounddown(uint32_t x, uint32_t y) {
    return (x / y) * y;
}


bool PosixMmapSequentialWriter::unmap() {

    if(m_base == NULL) return true;

    bool result = true;

    if (m_lastSync < m_mapSize) {
        // Defer syncing this data until next Sync() call, if any
        m_shouldSync = true;
    }
    if (munmap(m_base, m_mapSize) != 0) {
        result = false;
    }

    m_fileOffset += m_mapSize;
    m_base = NULL;
    m_lastSync = 0;
    m_dst = 0;

    return result;
}

bool PosixMmapSequentialWriter::map() {

    assert(m_base == NULL);
    assert(m_fd > 0);

    if (ftruncate(m_fd, m_fileOffset + m_mapSize) < 0) {
        return false;
    }
    void* buf = mmap(NULL, m_mapSize, PROT_READ | PROT_WRITE, MAP_SHARED, m_fd, m_fileOffset);
    if (buf == MAP_FAILED) {
        return false;
    }

    m_base = (char*)(buf);
    m_lastSync = 0;
    return true;
}

//append to the end of the file;
PosixMmapSequentialWriter::PosixMmapSequentialWriter(const std::string& fname,
                                                     uint32_t page_size, 
                                                     bool oscache)
: m_filename(fname),
  m_fd(-1),
  m_pageSize(page_size),
  m_mapSize(rounddown(1024*1024, page_size)),
  m_base(NULL),
  m_dst(0),
  m_lastSync(0),
  m_fileOffset(0),
  m_shouldSync(false),
  m_oscache(oscache) {
 
   assert((page_size & (page_size - 1)) == 0);
}    

//append to the specified offet
PosixMmapSequentialWriter::PosixMmapSequentialWriter(const std::string& fname,
                                                     uint32_t page_size, 
                                                     uint64_t offset, 
                                                     bool oscache)
: m_filename(fname),
  m_fd(-1),
  m_pageSize(page_size),
  m_mapSize(rounddown(1024*1024, page_size)),
  m_base(NULL),
  m_dst(0),
  m_lastSync(0),
  m_fileOffset(offset),
  m_shouldSync(false),
  m_oscache(oscache) {
 
   assert((page_size & (page_size - 1)) == 0);
}


PosixMmapSequentialWriter::~PosixMmapSequentialWriter() {

    if (m_fd >= 0) {
        PosixMmapSequentialWriter::close();
    }
}

DbStatus PosixMmapSequentialWriter::append(const char* buf, int len) {
 
     DbStatus ret;
     const char* src = buf;
	 int32_t left = len;
	 
     while (left > 0) {

       assert(m_dst <= m_mapSize);
	   int32_t avail = m_mapSize - m_dst;
	   if (m_base == NULL || avail == 0) {
		 if (!unmap() || !map()) {
		   return DbStatus(m_filename, errno);
		 }
	   }
	 
	   uint32_t n = (left <= avail) ? left : avail;
	   memcpy(m_base + m_dst, src, n);
	   m_dst += n;
	   src += n;
	   left -= n;
	 }

	 return ret;
 }

DbStatus PosixMmapSequentialWriter::append(const std::string& data) {
  
	return append(data.data(), data.size());
 }

DbStatus PosixMmapSequentialWriter::close() {
    assert(m_mapSize >= m_dst);

    DbStatus ret;
    if(m_base == NULL) return ret;

    if(m_fd < 0) return ret;

    uint32_t unused = m_mapSize - m_dst;
    if (!unmap()) {
        ret = DbStatus(m_filename, errno);
    } else if (unused > 0) {

        // Trim the extra space at the end of the file
        if (ftruncate(m_fd, m_fileOffset - unused) < 0) {
            ret = DbStatus(m_filename, errno);
        }
    }

    if(!m_oscache) {
        if (posix_fadvise(m_fd, m_pageSize, 0, POSIX_FADV_DONTNEED) < 0 ) ret = DbStatus(m_filename, errno);
    }

    if (::close(m_fd) < 0) {
        ret = DbStatus(m_filename, errno);
    }


    m_fd = -1;
    m_base = NULL;
    return ret;
}


DbStatus PosixMmapSequentialWriter::flush() {

    DbStatus ret;

    if(m_fd < 0 || m_base == NULL) return ret;

    if (m_shouldSync) {

        if(fsync(m_fd) < 0) {
            ret = DbStatus(m_filename, errno);
            return ret;
        }

        m_shouldSync = false;
    }

    if (m_dst > m_lastSync) {
        
        // Find the beginnings of the pages that contain the first and last
        // bytes to be synced.
        uint32_t p1 = rounddown(m_lastSync, m_pageSize);
        uint32_t p2 = rounddown(m_dst + m_pageSize - 1, m_pageSize);
        m_lastSync = m_dst;
        
        if (msync(m_base + p1, p2 - p1, MS_SYNC) < 0) {
            ret = DbStatus(m_filename, errno);
        }
    }

    return ret;
}

DbStatus PosixMmapSequentialWriter::init()
{
    DbStatus ret;

    m_fd = open(m_filename.c_str(), O_RDWR | O_CREAT, S_IRUSR|S_IWUSR); 
    if(m_fd < 0) return DbStatus("PosixMmapSequentialWriter::init " + m_filename, errno);

    uint64_t offset;
    ret = FsUtility::getFileSize(m_filename, offset);
    if(!ret.isok()) return ret;

    int64_t realoffset = (m_fileOffset < offset)?offset:m_fileOffset;
    m_fileOffset = rounddown(realoffset, m_pageSize);
    m_dst += realoffset - m_fileOffset;

    return ret;
}

uint32_t PosixMmapSequentialWriter::getNextOffset() {

	return m_fileOffset + m_dst;
}


PosixRandomAccessFile::PosixRandomAccessFile(const std::string& fname)
: m_filename(fname) {

}

PosixRandomAccessFile::~PosixRandomAccessFile() {

	if (m_fd >= 0) {
        PosixRandomAccessFile::close();
	}
}


DbStatus PosixRandomAccessFile::init() {

    DbStatus ret;

    m_fd = open(m_filename.c_str(), O_RDWR | O_CREAT, S_IRUSR|S_IWUSR); 
	if(m_fd < 0) return DbStatus(m_filename, errno);

    return ret;
}


DbStatus PosixRandomAccessFile::read(uint64_t offset, uint32_t n, std::string& result) {

    DbStatus ret;

	result.assign(n, 0);
	int r = pread(m_fd, (char*)result.c_str(), n, offset);
	if(r < 0) ret = DbStatus("PosixRandomAccessFile::read", errno);
	else if(r < n) {
    
        result.resize(r); 
    }
	return ret;
}


DbStatus PosixRandomAccessFile::read(uint64_t offset, uint32_t n, char* buf, uint32_t buflen, uint32_t& dlen) {

    DbStatus ret;

    if(n > buflen) return DbStatus("PosixRandomAccessFile::read", DbStatus::DB_BUFFERWILLOVERFLOW);

    int r = pread(m_fd, buf, n, offset);
    if(r < 0) ret = DbStatus("PosixRandomAccessFile::read", errno);
    dlen = r;

    return ret;
}


DbStatus PosixRandomAccessFile::write(uint64_t offset, const std::string& data) {

	return write(offset, data.data(), data.size());
}


DbStatus PosixRandomAccessFile::write(uint64_t offset, const char* buf, uint32_t n) {

    DbStatus ret;
	
    int rc = pwrite(m_fd, buf, n, offset);
	if(rc < 0) return  DbStatus("PosixRandomAccessFile::write", errno);

	return ret;
}


DbStatus PosixRandomAccessFile::close() {

    DbStatus ret;

    int rc = ::close(m_fd);
    if(rc < 0) return  DbStatus("PosixRandomAccessFile::close", errno);

    return ret;
}


DbStatus PosixRandomAccessFile::flush() {

    DbStatus ret;

    if (fsync(m_fd) < 0) {
        return DbStatus(m_filename, errno);
    }

    return ret;
}


DbStatus PosixRandomAccessFile::dropCache() {

    DbStatus ret;

    if (posix_fadvise(m_fd, 0, 0, POSIX_FADV_DONTNEED) < 0 ) ret = DbStatus(m_filename, errno); 

    return ret;
}


PosixMmapRandomAccessFile::PosixMmapRandomAccessFile(const std::string& fname, uint32_t mapSize)
: m_filename(fname), 
  m_fd(-1), 
  m_mapSize(mapSize), 
  m_base(NULL) {

}

PosixMmapRandomAccessFile::~PosixMmapRandomAccessFile() {

    if (m_fd >= 0) {
        PosixMmapRandomAccessFile::close();
    }

}


DbStatus PosixMmapRandomAccessFile::init() {

    DbStatus ret;

    m_fd = open(m_filename.c_str(), O_RDWR | O_CREAT, S_IRUSR|S_IWUSR);
    if(m_fd < 0) return DbStatus(m_filename, errno);

   
    uint64_t fileSize = 0;
    ret = FsUtility::getFileSize(m_filename, fileSize);
    if(!ret.isok() && ENOENT != ret.getCode()) return ret;

    if(fileSize < m_mapSize) {
        if (ftruncate(m_fd, m_mapSize) < 0) {
            return DbStatus(m_filename, errno);
        }
    }


    m_base = (char*)mmap(NULL, m_mapSize, PROT_READ | PROT_WRITE, MAP_SHARED, m_fd, 0);
    if (m_base == MAP_FAILED) {
        return DbStatus(m_filename, errno);
    }

    return ret;
}


DbStatus PosixMmapRandomAccessFile::read(uint64_t offset, uint32_t n,
        std::string& result) {

    DbStatus ret;

    result.clear();
    int len = (n + offset < m_mapSize)?n:(m_mapSize - offset);
    result.append(m_base + offset, len);
    
    return ret;
}


DbStatus PosixMmapRandomAccessFile::read(uint64_t offset, uint32_t n, char* buf, uint32_t buflen, uint32_t& dlen) {

    DbStatus ret;

    if(n > buflen) return DbStatus("PosixMmapRandomAccessFile::read", DbStatus::DB_BUFFERWILLOVERFLOW);

    dlen = (n + offset < m_mapSize)?n:(m_mapSize - offset);
    memcpy(buf, m_base + offset, dlen);

    return ret;
}


DbStatus PosixMmapRandomAccessFile::write(uint64_t offset, const std::string& data) {

    return write(offset, data.data(), data.size());
}


DbStatus PosixMmapRandomAccessFile::write(uint64_t offset, const char* buf, uint32_t n) {

    DbStatus ret;

    if(offset + n > m_mapSize) return DbStatus(m_filename, DbStatus::DB_BUFFERWILLOVERFLOW);
    memcpy(m_base + offset, buf, n);

    return ret;
}


DbStatus PosixMmapRandomAccessFile::close() {

    DbStatus ret;

    int rc = ::close(m_fd);
    if(rc < 0) return  DbStatus(m_filename, errno);

    if(munmap(m_base, m_mapSize) != 0) {
        return DbStatus(m_filename, errno); 
    }

    m_fd = -1; m_base = NULL;
    return ret;
}


DbStatus PosixMmapRandomAccessFile::flush() {

    DbStatus ret;

    if (msync(m_base, m_mapSize, MS_SYNC) < 0) {
       ret = DbStatus(m_filename, errno);
    }
    return ret;
}


DbStatus PosixMmapRandomAccessFile::dropCache() {

    DbStatus ret;

    return ret;
}
